package edu.cmu.vlis.wassup.sender;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.PrintWriter;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;

import sun.misc.Signal;

import com.sun.messaging.Topic;

import edu.cmu.vlis.wassup.IProcess;


public class MessageProcessor implements IProcess,Runnable{

    //Publisher one
    List<TopicSession> topicSessionList;
    List<TopicConnection> topicConnList;	
    Topic topic;
    private int _port;
    private ServerSocket socket;
    private Socket clientSocket;
    private PrintWriter out;
    private BufferedReader reader;
    private String[] urls;
    private Thread pollerThread;
    private boolean goIn;

    public MessageProcessor() {
        topicSessionList = new ArrayList<TopicSession>();
        topicConnList = new ArrayList<TopicConnection>();
        loadProperties();
        this.urls = System.getProperty("pubUrls").split(",");
        //Register hooker method
        Runtime.getRuntime().addShutdownHook(new
                ShutdownMessageProcessor(topicConnList));
        goIn = false;
        
        
    }

    @Override
    public void start(String name){
    	System.out.println("Initialising the Topic queues");
        try {
            // this needs to be corrected; there are multiple topics a MP connects to; not one.
            topic = new Topic(name.toLowerCase()); 
        } catch (JMSException e) {                                                           
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
        this.pollerThread = new Thread(this);
        this.pollerThread.start();
        
        Thread mpc = new Thread(new MPCoordinatorResponder(this.getID(),this));
        mpc.start();
    }

    public int getID(){
        return 30000;
    }

    @Override
    public void run() {
        this.startMP();
    }

    protected void reset(String[] newUrls){
        
        this.stopMP();
        this.urls = newUrls;

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        this.startMP();
    }
    
    public void startMP(){
        //Point to all publisher Topic Queue.
        //currently just one..
        if(topicConnList.isEmpty()) {
            
            TopicQueueFetcher tl = new TopicQueueFetcher(topic);
            for(int i=0;i<this.urls.length;i++) {
                try {
                    TopicConnectionFactory topicConnFactory = new com.sun.messaging.TopicConnectionFactory();

                    ((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
                            com.sun.messaging.ConnectionConfiguration.imqAddressList, this.urls[i] +  ":7676");
                    ((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
                            com.sun.messaging.ConnectionConfiguration.imqDefaultUsername, "admin");
                    ((com.sun.messaging.TopicConnectionFactory) topicConnFactory).setProperty(
                            com.sun.messaging.ConnectionConfiguration.imqDefaultPassword, "admin");

                    TopicConnection topicConn= topicConnFactory.createTopicConnection();

                    TopicSession topicSession = topicConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);

                    TopicSubscriber topicSubsciber = topicSession.createSubscriber(topic);
                    topicSubsciber.setMessageListener(tl);			
                    topicSessionList.add(topicSession);
                    topicConnList.add(topicConn);
                } catch(Exception je) {
                    System.out.println(je.toString());
                }
            }
        }
        for(TopicConnection topicConn : topicConnList) {
            try {
                topicConn.start();
            } catch (JMSException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }	
        }
    }

    public void stopMP() {
        //stop all the connection
        System.out.println("Stopping all the connections for Message Processor");
        for(TopicConnection topicConn : topicConnList) {
            try {
                if (topicConn != null)
                    topicConn.stop();
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

    public void shutDown() {
        //Sending kill stop signal to the process.
        Signal.raise(new Signal("INT"));
    }

    public void loadProperties() {
        Properties properties = new Properties();
        try {
            properties = new Properties();
            FileInputStream in = new FileInputStream("wassup/src/properties");
            properties.load(in);
            in.close();
            System.getProperties().putAll(properties);
        } catch (FileNotFoundException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        } catch (IOException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
}

